package com.bw.gmall.realtime.dwd.db.app;

import com.bw.gmall.realtime.common.base.BaseSqlApp;
import com.bw.gmall.realtime.common.constant.Constant;
import com.bw.gmall.realtime.common.util.SQLUtil;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.TableResult;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

/**
 * @Author sw
 * @CreateTime 2024-11-17
 */
public class DwdTradeRefundPaySucDetail extends BaseSqlApp {
    public static void main(String[] args) {
        new DwdTradeRefundPaySucDetail().start(Constant.TOPIC_DWD_TRADE_REFUND_PAYMENT_SUCCESS,4,10018);
    }


    @Override
    public void handle(StreamExecutionEnvironment env, StreamTableEnvironment tableEnv, String groupId) {
        //1、读取ods
        readOdsDb(tableEnv,groupId);
        //2、读取字典表
        createBaseDic(tableEnv);
        //3、过滤退款成功表数据
        getRefundPaymentTable(tableEnv);
        //4、过滤退单表中的退单成功的数据
        getOrderRefundInfoTable(tableEnv);
        //5、过滤订单表中的退款成功的数据
        getOrderInfoTable(tableEnv);
        //6、join
        Table result = getResult(tableEnv);
        //7、写到kafka
        exter(tableEnv);
        result.insertInto(Constant.TOPIC_DWD_TRADE_REFUND_PAYMENT_SUCCESS).execute();

    }

    private static TableResult exter(StreamTableEnvironment tableEnv) {
        return tableEnv.executeSql("create table " + Constant.TOPIC_DWD_TRADE_REFUND_PAYMENT_SUCCESS + "(" +
                "id string," +
                "user_id string," +
                "order_id string," +
                "sku_id string," +
                "province_id string," +
                "payment_type_code string," +
                "payment_type_name string," +
                "date_id string," +
                "callback_time string," +
                "refund_num string," +
                "refund_amount string," +
                "ts bigint " +
                ")" + SQLUtil.getKafkaSinkSQL(Constant.TOPIC_DWD_TRADE_REFUND_PAYMENT_SUCCESS));
    }

    private static Table getResult(StreamTableEnvironment tableEnv) {
        return tableEnv.sqlQuery(" select\n" +
                " rp.id,\n" +
                " oi.user_id,\n" +
                " rp.order_id,\n" +
                " rp.sku_id,\n" +
                " oi.province_id,\n" +
                " rp.payment_type,\n" +
                " c.info.dic_name payment_type_name,\n" +
                " date_format(rp.callback_time,'yyyy-MM-dd') date_id,\n" +
                " rp.callback_time,\n" +
                " ri.refund_num,\n" +
                " rp.total_amount,\n" +
                " rp.ts\n" +
                " from refund_payment rp\n" +
                " join order_refund_info ri\n" +
                " on rp.order_id = ri.order_id and rp.sku_id = ri.sku_id\n" +
                " join order_info oi\n" +
                " on rp.order_id = oi.id\n" +
                " JOIN base_dic FOR SYSTEM_TIME AS OF rp.proc_time AS c\n" +
                " ON rp.payment_type = c.rowkey");
    }

    private static void getOrderInfoTable(StreamTableEnvironment tableEnv) {
        Table table = tableEnv.sqlQuery("select " +
                "data['id'] id," +
                "data['user_id'] user_id," +
                "data['province_id'] province_id " +
                "from topic_db " +
                "where `database`='gmall_realtime' " +
                "and `table`='order_info' " +
                "and `type`='update' " +
                "and `old`['order_status'] is not null " +
                "and `data`['order_status']='1006'");
        tableEnv.createTemporaryView("order_info", table);
    }

    private static void getOrderRefundInfoTable(StreamTableEnvironment tableEnv) {
        Table table = tableEnv.sqlQuery("select " +
                "data['order_id'] order_id," +
                "data['sku_id'] sku_id," +
                "data['refund_num'] refund_num " +
                "from topic_db " +
                "where `database`='gmall_realtime' " +
                "and `table`='order_refund_info' " +
                "and `type`='update' " +
                "and `old`['refund_status'] is not null " +
                "and `data`['refund_status']='0705'");
        tableEnv.createTemporaryView("order_refund_info",table);
    }

    private static void getRefundPaymentTable(StreamTableEnvironment tableEnv) {
        Table table = tableEnv.sqlQuery("select " +
                "data['id'] id," +
                "data['order_id'] order_id," +
                "data['sku_id'] sku_id," +
                "data['payment_type'] payment_type," +
                "data['callback_time'] callback_time," +
                "data['total_amount'] total_amount," +
                "proc_time, " +
                "ts " +
                "from topic_db " +
                "where `database`='gmall_realtime' " +
                "and `table`='refund_payment' " +
                "and `type`='update' " +
                "and `old`['refund_status'] is not null " +
                "and `data`['refund_status']='1602'"
        );
        tableEnv.createTemporaryView("refund_payment",table);
    }
}
